/*******************************************************************************
 * Package: com.song.bigdata.stream
 * Type:    MyRichMapFunction
 * Date:    2022-10-30 13:55
 *
 *
 *
 * You may not use this file except in compliance with the License.
 *******************************************************************************/
package com.song.bigdata.stream;

import com.song.bigdata.stream.pojo.User;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 功能描述：
 *   富合函数练习
 * @author Songxianyang
 * @date 2022-10-30 13:55
 */
public class MyRichMapFunctionMain {
    public static void main(String[] args) throws Exception{
        // 创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        // 并行度
        environment.setParallelism(1);
        DataStreamSource<User> streamSource = environment.fromElements(
                new User("song-user", 12),
                new User("xian-user", 92),
                new User("song-user", 11),
                new User("song-user", 110),
                new User("song-user", 90),
                new User("xian-user", 10)
        );
        streamSource.map(new MyRichMapFunction()).print();
        environment.execute();
    }
}

class MyRichMapFunction extends RichMapFunction<User,Integer> {

    @Override
    public void open(Configuration parameters) throws Exception {

        // 获取当前运行时环境
        System.out.println("打开当前运行时"+getIterationRuntimeContext().getIndexOfThisSubtask()+"索引");
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        // 获取当前运行时环境
        System.out.println("关闭"+getIterationRuntimeContext().getIndexOfThisSubtask()+"索引");
        super.close();
    }


    @Override
    public Integer map(User user) throws Exception {
        return user.age;
    }
}
